package com.ssrs.batch.service.impl;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.log.Log;
import cn.hutool.log.LogFactory;
import com.ssrs.batch.core.callback.BatchJobCallback;
import com.ssrs.batch.core.exception.BatchServiceException;
import com.ssrs.batch.service.IBatchCreateReconcileService;
import com.ssrs.framework.util.SpringUtil;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;

/**
 * 批处理公共类
 *
 * @author ssrs
 */
@SuppressWarnings({"ALL", "AlibabaAvoidManuallyCreateThread"})
@Service
public class BatchCreateReconcileServiceImpl implements IBatchCreateReconcileService {
    private Log log = LogFactory.get(BatchCreateReconcileServiceImpl.class);

    @Autowired
    private Executor threadPoolTaskExecutor;

    @Autowired
    private JobLauncher jobLauncher;

    @Override
    public boolean run(String jobName, Map<String, Object> params) throws BatchServiceException {
        return run(jobName, params, false, null);
    }

    @Override
    public boolean run(String jobName, Map<String, Object> params, BatchJobCallback callback) throws BatchServiceException {
        return run(jobName, params, false, callback);
    }

    @Override
    public boolean run(String jobName, Map<String, Object> params, boolean isUseThreadPool) throws BatchServiceException {
        return run(jobName, params, isUseThreadPool, null);
    }

    @Override
    public boolean run(String jobName, Map<String, Object> params, boolean isUseThreadPool, BatchJobCallback callback) throws BatchServiceException {
        log.info("开始执行任务，jobName：{}", new Object[]{jobName});

        if (StrUtil.isBlank(jobName)) {
            return false;
        }

        try {
            //使用线程池管理
            if (isUseThreadPool) {
                // 开始配置信息
                if (callback != null) {
                    callback.beforeSetup();
                }

                // 执行运行
                threadPoolTaskExecutor.execute(getNamedThread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            call((Job) SpringUtil.getBean(jobName), params, callback);
                        } catch (BeansException e) {
                            log.error("创建任务失败", e);
                        } catch (BatchServiceException e) {
                            e.printStackTrace();
                            if (callback != null) {
                                callback.exception(e);
                            }

                        } finally {

                        }
                    }
                }));
                return true;
            } else {
                try {
                    return call((Job) SpringUtil.getBean(jobName), params, callback);
                } catch (BeansException e) {
                    log.error("创建任务失败", e);
                    return false;
                } catch (BatchServiceException e) {
                    e.printStackTrace();
                    if (callback != null) {
                        callback.exception(e);
                    }
                    return false;
                } finally {

                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("执行任务失败", e);
            return false;
        }
    }


    private boolean call(Job job, Map<String, Object> params, BatchJobCallback callback) throws BatchServiceException {
        JobExecution result = null;
        try {
            // 初始化数据
            if (callback != null) {
                callback.setup();
            }

            // 开始执行任务
            JobParametersBuilder builder = getJobParamsBuilder(params);
            result = jobLauncher.run(job, builder.toJobParameters());
        } catch (JobExecutionAlreadyRunningException e) {
            log.error("执行job失败，job名称：{}", new Object[]{job.getName()}, e);
            throw new BatchServiceException(ExitStatus.EXECUTING.getExitCode(), "任务正在执行, 请无重复执行");
        } catch (JobInstanceAlreadyCompleteException e) {
            log.error("执行job失败，job名称：{}", new Object[]{job.getName()}, e);
            throw new BatchServiceException(ExitStatus.COMPLETED.getExitCode(), "任务已完成, 请无重复执行");
        } catch (Exception e) {
            log.error("执行job失败，job名称：{}", new Object[]{job.getName()}, e);
            throw new BatchServiceException(ExitStatus.FAILED.getExitCode(), "执行任务失败:[" + e.getMessage() + "]");
        }

        ExitStatus es = result.getExitStatus();

        // 调用JOB回调函数
        if (callback != null) {
            String[] messages = createMessages(result);
            if (messages.length > 0) {
                callback.finished(es, messages);
            } else {
                callback.finished(es);
            }
        }

        String exitCode = es.getExitCode();
        if (ExitStatus.COMPLETED.getExitCode().equals(exitCode)) {
            log.info("任务执行完成，job名称：{}；exitCode={}；exitDesc={}", new Object[]{job.getName(), exitCode, es.getExitDescription()});
            return true;
        } else {
            log.debug("任务执行失败，job名称：{}；exitCode={}；exitDesc={}", new Object[]{job.getName(), exitCode, es.getExitDescription()});
            return false;
        }
    }

    /**
     * 创建Job参数创建对象
     *
     * @param params
     * @return
     * @throws ParseException
     */
    private JobParametersBuilder getJobParamsBuilder(Map<String, Object> params) throws ParseException {
        JobParametersBuilder builder = new JobParametersBuilder();
        if (params != null) {
            params.forEach((k, v) ->
            {
                if (v != null) {
                    if (v instanceof Integer) {
                        builder.addLong(k, ((Integer) v).longValue());
                    } else if (v instanceof Long) {
                        builder.addLong(k, (Long) v);
                    } else if (v instanceof String) {
                        builder.addString(k, (String) v);
                    } else if (v instanceof Double) {
                        builder.addDouble(k, (Double) v);
                    } else if (v instanceof Date) {
                        builder.addDate(k, (Date) v);
                    } else {
                        builder.addString(k, v.toString());
                    }
                }
            });
        }
        return builder;
    }

    /**
     * 获取线程并setName
     *
     * @param command
     * @return
     */
    @SuppressWarnings("AlibabaAvoidManuallyCreateThread")
    public Thread getNamedThread(Runnable command) {
        //noinspection AlibabaAvoidManuallyCreateThread
        @SuppressWarnings("AlibabaAvoidManuallyCreateThread") Thread thread = new Thread(command);
        // 设置线程name为"Clear."+方法名
        thread.setName("Clear." + Thread.currentThread().getStackTrace()[2].getMethodName());
        return thread;
    }

    /**
     * 创建消息内容
     *
     * @param jobExecution
     * @return
     */
    private String[] createMessages(JobExecution jobExecution) {
        List<String> messages = new ArrayList<>();
        List<Throwable> exceptions = jobExecution.getAllFailureExceptions();
        if (exceptions.size() > 0) {
            StringBuilder content = new StringBuilder();
            content.append("Job execution #");
            content.append(jobExecution.getId());
            content.append(" of job instance #");
            content.append(jobExecution.getJobInstance().getId());
            content.append(" failed with following exceptions:");
            messages.add(content.toString());
            for (Throwable exception : exceptions) {
                if (exception instanceof BatchServiceException) {
                    messages.add(ExceptionUtil.getMessage(exception)/*formatExceptionMessage(exception)*/);
                } else {
                    try {
                        if (exception.getCause().getCause() != null) {
                            messages.add(ExceptionUtil.getMessage(exception.getCause().getCause()));
                        }
                    } catch (Exception ex) {
                        messages.add(exception.getCause().getMessage());
                    }
                }
            }
        }
        return messages.toArray(new String[messages.size()]);
    }

    /**
     * 格式化异常消息
     *
     * @param exception
     * @return
     */
    private String formatExceptionMessage(Throwable exception) {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        exception.printStackTrace(new PrintStream(baos));
        return baos.toString();
    }
}
